package cn.org.intelli.zjgflink.service;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

@Service
public class HdfsOpService {
    private static final Logger log = LoggerFactory.getLogger(cn.org.intelli.zjgflink.service.HdfsOpService.class);

    @Value("${flink.hdfs.home}")
    private String flinkHdfsHome;

    public void deleteFileOrDirectory(String filePath) throws Exception {
        log.info("@DeleteFileOrDirectory: start to delete {}", filePath);
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        try {
            Path toDelete = new Path(filePath);
            Boolean isDeleted = Boolean.valueOf(fs.delete(toDelete, true));
            if (isDeleted.booleanValue()) {
                log.info("@DeleteFileOrDirectory: {} deleted successfully!", filePath);
            } else {
                log.info("@DeleteFileOrDirectory: {} deleted failed", filePath);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
    }

    public Boolean checkFolderExisted(String folderPath) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        try {
            return Boolean.valueOf(fs.exists(new Path(folderPath)));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
        return Boolean.valueOf(false);
    }

    public List<String> getFileNamesInFolder(String folderPath) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        List<String> fileNames = new ArrayList<>();
        try {
            FileStatus[] fileStatuses = fs.listStatus(new Path(folderPath));
            for (FileStatus fileStatus : fileStatuses)
                fileNames.add(fileStatus.getPath().getName());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
        return fileNames;
    }

    public FileStatus[] getFileStatusInFolder(String folderPath) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        List<String> fileNames = new ArrayList<>();
        try {
            FileStatus[] fileStatuses = fs.listStatus(new Path(folderPath));
            return fileStatuses;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            fs.close();
        }
        return null;
    }

    public void uploadFile2Hdfs(String localFilePath, String remoteFilePath) throws Exception {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);
        try {
            Boolean existed = Boolean.valueOf(fs.exists(new Path(remoteFilePath)));
            if (existed.booleanValue())
                throw new Exception("");
            FSDataOutputStream outputStream = fs.create(new Path(remoteFilePath));
            Files.copy(Paths.get(localFilePath), outputStream);
        } catch (Exception e) {
            e.printStackTrace();
            throw new Exception(e);
        } finally {
            fs.close();
        }
    }
}
